package drds.plus.sql_process.rule;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import drds.plus.common.Constants;
import drds.plus.common.lifecycle.AbstractLifecycle;
import drds.plus.common.model.Application;
import drds.plus.rule_engine.rule_calculate.DataNodeDataScatterInfo;
import drds.plus.sql_process.abstract_syntax_tree.configuration.TableMetaData;
import drds.plus.sql_process.abstract_syntax_tree.configuration.manager.RepositorySchemaManager;
import drds.plus.sql_process.abstract_syntax_tree.configuration.manager.SchemaManager;
import drds.plus.sql_process.abstract_syntax_tree.configuration.manager.StaticSchemaManager;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * 基于Rule获取到物理的group进行查找
 */
@Slf4j
public class RuleSchemaManager extends AbstractLifecycle implements SchemaManager {
    @Override
    public Logger log() {
        return log;
    }

    private final Application application;
    private RouteOptimizer routeOptimizer;
    private StaticSchemaManager staticSchemaManager;
    private boolean useCache = true;
    private LoadingCache<drds.plus.common.model.DataNode, RepositorySchemaManager> dataNodeToRepositorySchemaManagerCache = null;
    private LoadingCache<String, TableMetaData> tableNameToTableMetaDataCache = null;

    /**
     * default tableNameToTableMetaDataCache expire time, 30000ms
     */
    private long cacheExpireTime = Constants.DEFAULT_TABLE_META_EXPIRE_TIME;

    public RuleSchemaManager(RouteOptimizer routeOptimizer, Application application) {
        this(routeOptimizer, application, null);
    }

    public RuleSchemaManager(RouteOptimizer routeOptimizer, Application application, Long cacheExpireTime) {
        this.routeOptimizer = routeOptimizer;
        this.application = application;

        if (cacheExpireTime != null && cacheExpireTime != 0) {
            this.cacheExpireTime = cacheExpireTime;
        }

    }

    protected void doInit() {
        super.doInit();

        if (staticSchemaManager != null) {
            staticSchemaManager.init();
        }
        dataNodeToRepositorySchemaManagerCache = CacheBuilder.newBuilder().build(new CacheLoader<drds.plus.common.model.DataNode, RepositorySchemaManager>() {

            public RepositorySchemaManager load(drds.plus.common.model.DataNode dataNode) throws Exception {
                RepositorySchemaManager repositorySchemaManager = new RepositorySchemaManager();
                repositorySchemaManager.setDataNode(dataNode);
                repositorySchemaManager.setLocal(staticSchemaManager);
                repositorySchemaManager.setRouteOptimizer(routeOptimizer);
                repositorySchemaManager.init();
                return repositorySchemaManager;
            }
        });

        tableNameToTableMetaDataCache = CacheBuilder.newBuilder().maximumSize(1000).expireAfterWrite(cacheExpireTime, TimeUnit.MILLISECONDS).build(new CacheLoader<String, TableMetaData>() {

            public TableMetaData load(String tableName) throws Exception {
                return getTable0(tableName);

            }
        });
    }

    protected void doDestroy() {
        super.doDestroy();

        for (RepositorySchemaManager repositorySchemaManager : dataNodeToRepositorySchemaManagerCache.asMap().values()) {
            repositorySchemaManager.destroy();
        }

        if (tableNameToTableMetaDataCache != null) {
            tableNameToTableMetaDataCache.cleanUp();
        }
    }

    private TableMetaData getTable0(String tableName) {
        DataNodeDataScatterInfo dataNodeDataScatterInfo = routeOptimizer.shardAny(tableName);
        TableMetaData tableMetaData = null;
        if (dataNodeDataScatterInfo.getDataNodeId() == null) {
            // 没有对应的规则，也没有default datanode，则可能是一个不存在的表
            // 尝试找一下local
            tableMetaData = staticSchemaManager.getTableMetaData(tableName);
        } else {
            drds.plus.common.model.DataNode dataNode = application.getDataNode(dataNodeDataScatterInfo.getDataNodeId()); // 先找到group
            if (dataNode == null) {
                throw new RuntimeException("not found groupName : " + dataNodeDataScatterInfo.getDataNodeId());
            }
            try {
                tableMetaData = dataNodeToRepositorySchemaManagerCache.get(dataNode).getTable(tableName, dataNodeDataScatterInfo.getTableNameSet().iterator().next());
            } catch (ExecutionException e) {
                log.error(("not found where : " + tableName), e);
            }
        }

        return tableMetaData;
    }

    public TableMetaData getTableMetaData(String tableName) {
        // if (tableName.equals(DUAL)) {
        // return buildDualTable();
        // }

        TableMetaData tableMetaData = null;
        if (staticSchemaManager != null) {// 本地如果开启了，先找本地
            tableMetaData = staticSchemaManager.getTableMetaData(tableName);
        }
        if (tableMetaData != null) {
            return tableMetaData;
        }
        if (useCache) {
            try {
                tableMetaData = tableNameToTableMetaDataCache.get(tableName);
            } catch (Throwable e) {
                throw new RuntimeException("not found where : " + tableName);
            }
        } else {
            tableMetaData = this.getTable0(tableName);
        }
        return tableMetaData;
    }

    public void putTable(String tableName, TableMetaData tableMetaData) {
        if (staticSchemaManager != null) {
            staticSchemaManager.putTable(tableName, tableMetaData);
        } else if (useCache) {
            tableNameToTableMetaDataCache.put(tableName, tableMetaData);
        }
    }

    public Collection<TableMetaData> getAllTables() {
        List<TableMetaData> tableMetaDataList = new ArrayList();
        if (staticSchemaManager != null) {
            tableMetaDataList.addAll(staticSchemaManager.getAllTables());
        }

        if (tableNameToTableMetaDataCache != null) {
            tableMetaDataList.addAll(tableNameToTableMetaDataCache.asMap().values());
        }

        return tableMetaDataList;
    }


    public void setRouteOptimizer(RouteOptimizer routeOptimizer) {
        this.routeOptimizer = routeOptimizer;
    }

    public void setStaticSchemaManager(StaticSchemaManager staticSchemaManager) {
        this.staticSchemaManager = staticSchemaManager;
    }

    public void setUseCache(boolean useCache) {
        this.useCache = useCache;
    }

    public void reload() {
        this.tableNameToTableMetaDataCache.invalidateAll();
    }

}
